package com.lefu.risk.storm.bolt;


import com.lefu.risk.storm.common.constants.Constant;
import com.lefu.risk.storm.to.OrderDataTo;
import com.lefu.risk.storm.utils.JedisPoolUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

/**
 * Created by mwang on 2016/12/29.
 */
public class OrderReportBolt extends BaseRichBolt {
    private Map<String,Double> commonMap = null;
    private Map<String,Double> largestDayAmountMap = null;

    private JedisPoolUtil jedisPoolUtil = null;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        jedisPoolUtil = new JedisPoolUtil(Constant.REDIS_PROPERTIES_NAME);

    }

    @Override
    public void execute(Tuple input) {
        try{
            long orderDataBoltTime = input.getLongByField("orderDataBoltTime");
            OrderDataTo orderDataBoltTo = (OrderDataTo)input.getValueByField("orderDataTo");
            commonMap = (Map<String, Double>) orderDataBoltTo.getCommonMap();
            largestDayAmountMap = (Map<String, Double>) orderDataBoltTo.getLargestDayAmountMap();
            resultToRedis(commonMap,largestDayAmountMap);
            jedisPoolUtil.hset(Constant.ORDER_DATA_TIME_CONSUMING_REDIS_KEY,input.getStringByField("id"),("" + ((Long.parseLong(jedisPoolUtil.time().get(0)) * Constant.SECOND_TO_MICROSECONDS + Long.parseLong(jedisPoolUtil.time().get(1))) - orderDataBoltTime)/1000));

        }catch (Exception e){
            e.printStackTrace();
        }
//        finally {
//            if(jedisPoolUtil != null){
//                jedisPoolUtil.close();
//            }
//        }
    }

    private void resultToRedis(Map<String, Double> commonMap, Map<String, Double> largestDayAmountMap) {
        largestDayAmountMap.forEach((k,v) -> {
            jedisPoolUtil.hset(Constant.LARGEST_DAY_AMOUNT_MAP_KEY_MAX_AMOUNT,k,v.toString());
        });

        jedisPoolUtil.hset(Constant.TRANSAMOUNT_REDIS_KEY,Constant.COMMON_MAP_KEY_TRANS_AMOUNT_SUCCESS,commonMap.get(Constant.COMMON_MAP_KEY_TRANS_AMOUNT_SUCCESS).toString());
        jedisPoolUtil.hset(Constant.TRANSAMOUNT_REDIS_KEY,Constant.COMMON_MAP_KEY_TRANS_AMOUNT_FAILED,commonMap.get(Constant.COMMON_MAP_KEY_TRANS_AMOUNT_FAILED).toString());
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }

    @Override
    public void cleanup() {
        System.out.println("largestDayAmountMap.get(LARGEST_DAY_AMOUNT_MAP_KEY_MAX_AMOUNT)------> " + largestDayAmountMap.get(Constant.LARGEST_DAY_AMOUNT_MAP_KEY_MAX_AMOUNT));
        System.out.println("commonMap.get(COMMON_MAP_KEY_TRANS_AMOUNT_SUCCESS)------> " + commonMap.get(Constant.COMMON_MAP_KEY_TRANS_AMOUNT_SUCCESS));
        System.out.println("commonMap.get(COMMON_MAP_KEY_TRANS_AMOUNT_FAILED)------> " + commonMap.get(Constant.COMMON_MAP_KEY_TRANS_AMOUNT_FAILED));



    }
}
